package com.ydl.learning.flink.demo.transformations

import com.ydl.learning.flink.demo.Utils
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * apply是对window流做自定义的复杂运算用的
 *
 * @author ydl
 * @since 2020/11/25
 */
class ApplyDemo extends Utils with App {
  val stream = getSocketTextStream(9999)
  stream._2.flatMap(_.split(" ")).map((_, 1)).keyBy(_._1).timeWindow(Time.seconds(30))
    //通过apply实现reduce的功能
    .apply((k: String, w: TimeWindow, t: Iterable[(String, Int)], out: Collector[(String, Int)]) => {
      println(w.getStart)
      println(w.getEnd)
      var count: Int = 0
      for (elem <- t) {
        count += elem._2
      }
      out.collect((k, count))
    }
    )

}
